package com.tcm.kafka;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.Future;

public class ProducerCompanyStart {

    public static final String brokerList = "localhost:9092";
    public static final String topic = "topic-demo";

    public static Properties initConfig(){
        Properties properties = new Properties();
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
//        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, CompanySerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, CompanyProtostuffSerializer.class.getName());
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        properties.put(ProducerConfig.CLIENT_ID_CONFIG, "client-1");
        properties.put(ProducerConfig.RETRIES_CONFIG, 10);
        return properties;
    }

    public static void main(String[] args) {
//        Properties properties = new Properties();
//        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
//        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
//        properties.put("bootstrap.servers", brokerList);
        Properties properties = initConfig();

        //配置生产者客户端参数并创建KafkaProducer实例
        KafkaProducer<String, Company> producer = new KafkaProducer<String, Company>(properties);

        for(int i=0;i<1000;i++) {
            Company company = new Company("name"+i, "address"+i);
            //构建需要发送的消息
            ProducerRecord<String, Company> record = new ProducerRecord(topic, company);

            // 发送消息
            try {
                Future<RecordMetadata> future = producer.send(record, new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                        System.out.println(1 + ":" + recordMetadata.topic() + "->" + recordMetadata.partition() + ":" + recordMetadata.offset());
                    }
                });
                RecordMetadata metadata = future.get();
                System.out.println(2 + ":" + metadata.topic() + "->" + metadata.partition() + ":" + metadata.offset());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        //关闭生产者客户端示例
        producer.close();
    }

}
